package top.lzzly.sync.kafka.server;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @Author smq
 * @Date 2021/10/29 15:09
 * @Uint d9lab
 * @Description: ES 通用业务逻辑层
 */
@Slf4j
@Service
public class EsServiceImpl implements IEsService {

    @Autowired
    @Qualifier("restHighLevelClient")
    private RestHighLevelClient client;

    @Override
    public void addQuestionToEs(String id, String esType, Object object) throws IOException {
        IndexRequest request = new IndexRequest(esType);
        request.id(id);
        request.timeout(TimeValue.timeValueSeconds(1));
        request.source(JSON.toJSONString(object), XContentType.JSON);
        client.index(request, RequestOptions.DEFAULT);
    }

    @Override
    public void deleteQuestion(String id, String esType) {
        DeleteRequest request = new DeleteRequest(esType, id);
        request.timeout("1s");
        try {
            client.delete(request, RequestOptions.DEFAULT);
            log.info("删除es中的表：{}，数据：{}", esType, id);
        } catch (IOException e) {
            log.error("es中的数据删除失败，id为：{}", id);
        }
    }

    @Override
    public <T> List<T> search(String index, Map<String, String> map,Class<T> t) throws IOException {
        List<T> list = new ArrayList<>();
        SearchRequest request = new SearchRequest(index);
        //构建搜索条件
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        //查询条件，我们可以使用QueryBuilders 工具来实现
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        for (Iterator iter = map.entrySet().iterator(); iter.hasNext();) {
            Map.Entry entry = (Map.Entry) iter.next();
            boolQueryBuilder = QueryBuilders.boolQuery()
                    .must(QueryBuilders.matchQuery(entry.getKey().toString(), entry.getValue()));
        }
        sourceBuilder.query(boolQueryBuilder);
        sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
        request.source(sourceBuilder);
        SearchResponse searchResponse = client.search(request, RequestOptions.DEFAULT);
        Arrays.stream(searchResponse.getHits().getHits())
                .forEach(i -> {
                    String sourceAsString = i.getSourceAsString();
                    JSONObject resJson = (JSONObject) JSONObject.parse(sourceAsString);
                    T res = JSONObject.parseObject(String.valueOf(resJson), t);
                    list.add(res);
                });
        return list;
    }
}
